package com.winit.schedule.job.tracker.remoting;

import static com.winit.schedule.core.protocol.Protocol.RequestCode.*;
import io.netty.channel.ChannelHandlerContext;

import java.util.HashMap;
import java.util.Map;

import com.winit.schedule.core.protocol.AbstractCommandBody;
import com.winit.schedule.core.protocol.Protocol.RequestCode;
import com.winit.schedule.job.tracker.channel.ChannelManager;
import com.winit.schedule.job.tracker.channel.ChannelWrapper;
import com.winit.schedule.remoting.exception.RemotingCommandException;
import com.winit.schedule.remoting.netty.NettyRequestProcessor;
import com.winit.schedule.remoting.protocol.RemotingCommand;
import com.winit.schedule.remoting.protocol.RemotingProtos;

public class RemotingDispatcher extends AbstractProcessor {

    private static final Map<RequestCode, NettyRequestProcessor> processors = new HashMap<RequestCode, NettyRequestProcessor>();

    public RemotingDispatcher(RemotingService remotingService){
        super(remotingService);

        processors.put(HEART_BEAT, new HeartBeatProcessor(remotingService));
        processors.put(COMPLETE_TASK, new CompleteTaskProcessor(remotingService));
        processors.put(QUARTZ_REQ, new QuartzProcessor(remotingService));
    }

    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
                                                                                             throws RemotingCommandException {
        RequestCode code = RequestCode.valueOf(request.getCode());
        NettyRequestProcessor processor = processors.get(code);
        if (processor == null) {
            return RemotingCommand.createResponseCommand(RemotingProtos.ResponseCode.REQUEST_CODE_NOT_SUPPORTED.code(),
                "request code not supported!");
        }
        commonHandler(ctx, request);
        return processor.processRequest(ctx, request);
    }

    /**
     * 1. 将 channel 纳入管理中(不存在就加入) 2. 更新 TaskTracker 节点信息(可用线程数)
     * 
     * @param ctx
     * @param request
     */
    private void commonHandler(ChannelHandlerContext ctx, RemotingCommand request) {
        AbstractCommandBody commandBody = request.getBody();
        String taskTrackerUuid = commandBody.getIdentity();

        // 1. 将 channel 纳入管理中(不存在就加入)
        if (null != taskTrackerUuid) {
            ChannelManager.addChannel(new ChannelWrapper(ctx.channel(), taskTrackerUuid));
        }
    }
}
